package com.telra.app.server.jms;

import javax.jms.Connection;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.Topic;

import com.telra.app.server.cassandra.CassandraWriter;
import com.telra.app.server.data.OrderEvent;
import com.telra.app.server.protobuff.OrderProtos;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;

import com.espertech.esper.client.EPServiceProvider;


/**
 * Listens to orders messages from the JMS publisher and sends an OrderEvent to Esper
 * and to Cassandra to be queued for insert
 *
 * @author jamesarlet
 *
 */
public class Listener {

    private static final Log4JLogger log = (Log4JLogger) LogFactory.getFactory().getInstance(Listener.class);
    private EPServiceProvider serviceProvider;
    private CassandraWriter cassandra;

    public Listener(EPServiceProvider serviceProvider, CassandraWriter cassandra) {
        this.serviceProvider = serviceProvider;
        this.cassandra = cassandra;
        Thread brokerThread = new Thread(new ListenerThread());
        brokerThread.setName("listener");
        brokerThread.start();
    }


    private class ListenerThread implements Runnable, ExceptionListener {

        public void run() {

            try {

                // Create a ConnectionFactory
                ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616");

                // Create a Connection
                Connection connection = connectionFactory.createConnection();
                connection.start();

                connection.setExceptionListener(this);

                // Create a Session
                Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

                // Create the destination (Topic or Queue)
                Topic topic = session.createTopic("TEST.OUT");

                // Create a MessageConsumer from the Session to the Topic or Queue
                MessageConsumer consumer = session.createConsumer(topic);
                log.info("started listener");

                while (true) {
                    Message message = consumer.receive(1000);	//wait 1 second

                    if (message instanceof ObjectMessage) {
                        ObjectMessage bMessage = (ObjectMessage) message;
                        OrderProtos.OrderTicket o = (OrderProtos.OrderTicket)bMessage.getObject();
                        log.debug("recv: " + o.getItemName() + "," + o.getItemCategory() + "," + o.getPrice());

                        int catNum = o.getItemCategory().getNumber();

                        //{CLOTHING, ACCESSORIES, FOOTWEAR, OUTERWEAR, MISC};
                        OrderEvent.Category cat = (catNum==0?OrderEvent.Category.CLOTHING:
                                catNum==1?OrderEvent.Category.ACCESSORIES:
                                        catNum==2?OrderEvent.Category.FOOTWEAR:
                                                catNum==3?OrderEvent.Category.OUTERWEAR:
                                                        OrderEvent.Category.MISC);
                        OrderEvent order = new OrderEvent(o.getItemName(), cat, o.getPrice());
                        serviceProvider.getEPRuntime().sendEvent(order);	//send to Esper
                        cassandra.queueInsert(order);
                    }
                }

                //consumer.close();
                //session.close();
                //connection.close();
            } catch (Exception e) {
                log.error("Caught: " + e);
                e.printStackTrace();
            }

        }

        public void onException(JMSException e) {
            log.error("Caught: " + e);
            e.printStackTrace();
        }

    }


}
